package com.shujia.flink.transformation

import org.apache.flink.streaming.api.scala._
import org.apache.flink.api.common.functions.{FlatMapFunction, RichFlatMapFunction, RuntimeContext}
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.util.Collector

object Demo8SideOut {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val ds1: DataStream[String] = env.socketTextStream("master", 9999)

    val nan: OutputTag[String] = OutputTag[String]("男")
    val nv: OutputTag[String] = OutputTag[String]("女")


    val splitDS: DataStream[String] = ds1.process(new ProcessFunction[String, String] {
      override def processElement(value: String, ctx: ProcessFunction[String, String]#Context, out: Collector[String]): Unit = {
        val sex: String = value.split(",")(3)

        sex match {
          case "男" => ctx.output(nan, value)
          case "女" => ctx.output(nv, value)
          case _ => println("其它")
        }
      }
    })

    val nanDS: DataStream[String] = splitDS.getSideOutput(nan)

    val nvDS: DataStream[String] = splitDS.getSideOutput(nv)

    nanDS.print()


    env.execute()


  }
}
